package com.yujinglin.maxmin;


import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Mapper;
import org.jetbrains.annotations.NotNull;

import java.io.IOException;

/**
 * MaxMin多个值一次输出给Reduce时，考虑自定义Writable类型
 */
public class MaxMinMapper extends Mapper<LongWritable, Text, IntWritable,DoubleWritable> {

    private  double max;
    private  double min;

    private IntWritable outKey = new IntWritable();

    /**
     * 对于所有数字的计数
     */
    private  long count;
    /**
     * 只在Mapper对象被实例化时，执行一次
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {

        this.max=Long.MIN_VALUE;
        this.min =Long.MAX_VALUE;

    }

    /**
     * 每一行数据执行一次,对于求极值来说，map方法中，不需要执行context.write()
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, @NotNull Text value, Context context) throws IOException, InterruptedException {


        String line = value.toString();
        String[] info = line.split(",");

        int distance = 0;

        double speed = 0;

        if (info.length >= 19 && !info[4].isEmpty() && !info[17].isEmpty() && !info[5].isEmpty()&& !info[16].isEmpty() ) {

            try {
                distance = Integer.parseInt(info[5]) / 500;
                if(distance>5) distance=6;

                speed= Double.parseDouble(info[17]) / Double.parseDouble(info[16]);
            }catch (Exception e){
                e.printStackTrace();
            }


            outKey.set(distance); // 设置输出的key为距离分类
            context.write(outKey,new DoubleWritable(speed));
        }





//            // 判断是否是更大的值
//            if ( speed > max) {
//                max = speed;
//            }
//            if (speed < min) {
//                min = speed;
//            }
//
//        MaxMinWritable mm = new MaxMinWritable(max,min);


    }

//    @Override
//    protected void cleanup(Context context) throws IOException, InterruptedException {
//

//    }
}
